package com.shu.mq;

import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @author ssm
 * @version V1.0
 * @description:
 * @date 2020/9/14 14:38
 */

// 生产者 事务模式特别慢,所以不考虑,都是用confirm模式
@Slf4j
public class Producer {
    private static final String EXCHANGE_NAME = "exchange_return_1";
    // 可以路由的key
    private static final String ROUTING_KEY = "return.save";
    // 不能路由的key
    private static final String ROUTING_KEY_ERROR = "km.save";
    private static final String QUEUE_NAME = "queue_return_1";

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        Connection connection = ConnectionUtils.getConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, "topic");

        channel.confirmSelect();
        channel.addConfirmListener(new ConfirmListener() {
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                log.info("{},{} handleAck", deliveryTag, multiple);
            }

            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                log.info("{},{} handleNack", deliveryTag, multiple);
            }
        });


        // 添加return监听机制
        channel.addReturnListener(new ReturnListener() {
            /**
             *  监听的回调方法
             * @param replyCode 队列响应给浏览器的状态码
             * @param replyText 状态码对应的文本信息
             * @param exchange  交换机的名称
             * @param routingKey 路由的key
             * @param properties 消息的相关属性
             * @param body  消息体的内容
             */
            @Override
            public void handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("监听到不可达消息..." + new String(body));
                System.out.println("状态码：" + replyCode + "  文本信息为：" + replyText);
                System.out.println("交换机的名字为：" + exchange + "  路由的key为：" + routingKey);
                System.out.println("相关属性：" + properties);
            }
        });
        for (int i = 0; i < 10; i++) {
            if (i % 2 == 0) {
                // mandatory 属性设置为true表示：要监听不可达的消息
                channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, true, null, ("return监听机制" + i).getBytes());
            } else {
                channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY_ERROR, true, null, ("return监听机制" + i).getBytes());
            }
        }

        // 效率最高的还是异步监听方式

//        if (channel.waitForConfirms()) {
//            System.out.println("消息发送成功" );
//        }
//
//        channel.waitForConfirmsOrDie(); //直到所有信息都发布，只要有一个未确认就会IOException
//        System.out.println("全部执行完成");
    }
}